package com.gper.edu.flink.stream.window;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: ellis.guan
 * @Description: 增量聚合
 * @Date: 2020/11/8 13:15
 * @Version: 1.0
 */
public class MyAggregateFunction implements AggregateFunction<Tuple3<Integer, Double, Long>, Tuple2<Integer, Double>, Tuple2<Integer, Double>> {
    @Override
    public Tuple2<Integer, Double> createAccumulator() {
        return Tuple2.of(null, 0.0);
    }

    @Override
    public Tuple2<Integer, Double> add(Tuple3<Integer, Double, Long> input, Tuple2<Integer, Double> acc) {
        System.out.println("----aggregate value : "+String.format("%s : %s ", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(input.f2)),
                                         input.toString()));
        if (acc.f0 == null) {
            acc.f0 = input.f0;
        }
        acc.f1 += input.f1;
        return acc;
    }

    @Override
    public Tuple2<Integer, Double> getResult(Tuple2<Integer, Double> value) {
        return value;
    }

    @Override
    public Tuple2<Integer, Double> merge(Tuple2<Integer, Double> value, Tuple2<Integer, Double> acc1) {
        acc1.f1 += value.f1;
        return acc1;
    }
}